package com.xxd.dfs.client;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * @author: XiaoDong.Xie
 * @create: 2020-09-07 11:56
 * @description: 网络管理组件
 * <p>
 * 对每一个datanode (host) 去维护一个容器 host -> SelectionKey
 * 异步去改变 SelectionKey 的 感兴趣事件
 * 同一时间 只能对一个 datanode 发送一个数据请求
 */
public class NetworkManager {

    private static final Logger logger = LoggerFactory.getLogger(NetworkManager.class);

    // 正在连接中
    public static final Integer CONNECTING = 1;
    // 已经建立连接
    public static final Integer CONNECTED = 2;
    // 断开连接
    public static final Integer DISCONNECTED = 3;

    /**
     * 待连接主机队列
     */
    private ConcurrentLinkedQueue<Host> waitingConnectQueue = new ConcurrentLinkedQueue<>();

    /**
     * 主机对应的所有的等待发送队列
     */
    private Map<String, ConcurrentLinkedQueue<NetworkRequest>> hostWaitingSendQueue = new ConcurrentHashMap<>();

    /**
     * 当前主机正在发送的请求队列
     */
    private Map<String, NetworkRequest> toSendQueue = new ConcurrentHashMap<>();

    /**
     * 发送请求感兴趣的时间
     */
    private Map<String, SelectionKey> host2SelectKey = new ConcurrentHashMap<>();

    private Selector selector = null;

    /**
     * 客户端连接状态
     */
    private Map<String, Integer> connectState = new HashMap<>();


    public NetworkManager() {
        try {
            this.selector = Selector.open();
        } catch (IOException e) {
            e.printStackTrace();
        }

        new NetworkConnectThread().start();
    }

    /**
     * 连接到DataNode
     * 需要判断一下 当前客户端有没有连接过datanode 连接状态
     * 连接的组件
     *
     * @param host
     * @return
     */
    public boolean connectToDN(Host host) {
        String hostname = host.getHostname();
        if (!connectState.containsKey(hostname)
                || connectState.get(hostname).equals(DISCONNECTED)) {
            connectState.put(hostname, CONNECTING);
            waitingConnectQueue.offer(host);
        }

        while (connectState.get(hostname).equals(CONNECTING)) { // 正在连接
            try {
                wait(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        if (connectState.get(hostname).equals(DISCONNECTED)) { // 连接失败了
            logger.info("连接建立失败.......");
            return false;
        }

        return true;
    }

    /**
     * 发送请求
     *
     * @param request
     */
    public void sendRequest(NetworkRequest request) {
        ConcurrentLinkedQueue<NetworkRequest> queue = hostWaitingSendQueue.get(request.getHostname());
        queue.offer(request);
    }

    /**
     * 网络连接建立的核心线程
     */
    class NetworkConnectThread extends Thread {

        @Override
        public void run() {
            while (true) {
                tryConnect();
                prepareRequests();
                finishedConnect();
            }
        }

        /**
         * 尝试完成网络连接、请求发送、响应读取
         */
        private void finishedConnect() {
            SocketChannel channel = null;

            try {
                selector.select();
                Iterator<SelectionKey> keyIterator = selector.selectedKeys().iterator();
                while (keyIterator.hasNext()) {
                    SelectionKey key = keyIterator.next();
                    keyIterator.remove();

                    channel = (SocketChannel) key.channel();
                    if (key.isConnectable()) {
                        connectEvent(key, channel);
                    } else if (key.isWritable()) {
                        writeEvent(key, channel);
                    } else if (key.isReadable()) {
                        readEvent(key, channel);
                    }
                }
            } catch (IOException e) {
                logger.error("selector 出现异常", e);
                if (channel != null) {
                    try {
                        channel.close();
                    } catch (IOException e1) {
                        e1.printStackTrace();
                    }
                }
            }
        }

        /**
         * 读取响应数据事件
         * 主要要对几个容器进行清除 toSendQueue host2SelectionKey 等等
         *
         * @param key
         * @param channel
         */
        private void readEvent(SelectionKey key, SocketChannel channel) throws IOException {
            InetSocketAddress remoteAddress = (InetSocketAddress) channel.getRemoteAddress();
            String hostname = remoteAddress.getHostName();

            NetworkRequest request = toSendQueue.get(hostname);
            NetworkResponse response = null;

            // 不同的数据请求有不同的响应
            if (request.getRequestType().equals(NetworkRequest.REQUEST_SEND_FILE)) {
                response = getSendFileResponse(request.getId(), hostname, channel);
            } else if (request.getRequestType().equals(NetworkRequest.REQUEST_READ_FILE)) {
                response = getReadFileResponse(request.getId(), hostname, channel);
            }

            if (!response.getFinished()) { // 判断是否完成
                return;
            }

            // 取消监听key的事件
            key.interestOps(key.interestOps() & ~SelectionKey.OP_READ);

            if (request.getNeedResponse()) {
               // finishedResponses.put(request.getId(), response);
            } else {
                if (request.getCallback() != null) {
                    request.getCallback().process(response);
                }
            }

            toSendQueue.remove(hostname);
            host2SelectKey.remove(hostname);
        }

        /**
         * 获取下载文件的响应
         * @param requestId
         * @param hostname
         * @param channel
         * @return
         */
        private NetworkResponse getReadFileResponse(String requestId, String hostname, SocketChannel channel) {
            return null;
        }

        /**
         * 上传文件的雄鹰
         * @param requestId
         * @param hostname
         * @param channel
         * @return
         */
        private NetworkResponse getSendFileResponse(String requestId, String hostname, SocketChannel channel)
                throws IOException {
            NetworkResponse response = new NetworkResponse();
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            channel.read(buffer);
            buffer.flip();

            response.setRequestId(requestId);
            response.setHostname(hostname);
            response.setBuffer(buffer);
            response.setError(false);
            response.setFinished(true);

            return response;
        }

        /**
         * 发送数据事件
         *
         * @param key
         * @param channel
         */
        private void writeEvent(SelectionKey key, SocketChannel channel) {
            InetSocketAddress remoteAddress = null;
            try {
                remoteAddress = (InetSocketAddress) channel.getRemoteAddress();
                String hostName = remoteAddress.getHostName();

                // 获取要发送到这台机器的请求的数据
                NetworkRequest request = toSendQueue.get(hostName);
                ByteBuffer buffer = request.getBuffer();

                // 将数据发送出去
                // 注意 这里如果文件很大的话，不一定能一下子就直接给发送出去
                while (buffer.hasRemaining()) {
                    channel.write(buffer);
                }

                logger.info("本次发送请求至 datanode = {}，数据发送完毕.....", hostName);
                key.interestOps(SelectionKey.OP_READ);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        /**
         * 处理连接事件
         * 连接建立，默认就会发送数据，所以这里会把改hostname 初始化一个队列
         * 发送那些连接请求
         *
         * @param key
         * @param channel
         */
        private void connectEvent(SelectionKey key, SocketChannel channel) {
            InetSocketAddress remoteAddress = null;
            if (channel.isConnectionPending()) {
                try {
                    remoteAddress = (InetSocketAddress) channel.getRemoteAddress();
                    while (channel.finishConnect()) {
                        Thread.sleep(50);
                    }

                    logger.info("完成与{} : {}的连接 建立.....", remoteAddress.getHostName(), remoteAddress.getPort());
                    String hostName = remoteAddress.getHostName();
                    hostWaitingSendQueue.put(hostName, new ConcurrentLinkedQueue<>());
                    connectState.put(hostName, CONNECTED);
                    host2SelectKey.put(hostName, key);
                } catch (IOException e) {
                    logger.error("连接失败", e);
                    if (remoteAddress != null) {
                        connectState.put(remoteAddress.getHostName(), DISCONNECTED);
                    }
                } catch (InterruptedException e2) {
                    logger.error("thread error", e2);
                }
            }
        }

        /**
         * 准备好要发送的请求
         */
        private void prepareRequests() {
            for (String hostname : hostWaitingSendQueue.keySet()) {
                ConcurrentLinkedQueue<NetworkRequest> requestQueue = hostWaitingSendQueue.get(hostname);
                if (!requestQueue.isEmpty() && !toSendQueue.containsKey(hostname)) {
                    // 主机等待发送数据的请求 加入正在发送的队列
                    toSendQueue.put(hostname, requestQueue.poll());

                    // 准备好发送请求，应该对感兴趣的key 进行改变
                    SelectionKey key = host2SelectKey.get(hostname);
                    key.interestOps(SelectionKey.OP_WRITE);
                }
            }
        }

        /**
         * 尝试把排队中的机器连接起来
         */
        private void tryConnect() {
            SocketChannel channel = null;
            Host host = null;
            while ((host = waitingConnectQueue.poll()) != null) {
                try {
                    channel = SocketChannel.open();
                    channel.configureBlocking(false);
                    channel.connect(new InetSocketAddress(host.getHostname(), host.getNioPort()));
                    channel.register(selector, SelectionKey.OP_CONNECT);
                } catch (IOException e) {
                    logger.error("连接失败", e);
                    connectState.put(host.getHostname(), DISCONNECTED);
                }
            }
        }
    }
}
